package com.bw.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.bean.TrafficSourceStats;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyClickHouseUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

/**
 * 流量来源TOP10统计应用
 * 统计商品流量、店铺流量、直播流量、内容流量四大模块的TOP10来源
 * - TrafficSourceTopNApp.java ：流量来源TOP N统计
 */
/**
 * 流量域流量来源TopN应用
 * 负责统计网站流量来源渠道的热度排行，分析用户获取渠道效果
 * 主要功能点：
 * - 从Kafka的ods_traffic主题读取数据（注释提到实际应从dwd层读取）
 * - 进行JSON转换处理
 * - 提取流量来源渠道信息
 * - 统计各渠道的访问量
 * - 计算TopN流量来源渠道
 * 
 * 数据流向：
 * 流量数据 -> TrafficSourceTopNApp -> 处理后输出流量来源渠道TopN结果
 */
public class TrafficSourceTopNApp {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 2. 读取Kafka的页面日志主题
        // 注意：在实际环境中应该从dwd层读取，但为了测试可以直接从ods层读取
        String pageTopic = "ods_traffic";
        String groupId = "traffic_source_topn_group_test";
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getFlinkKafkaConsumer(pageTopic, groupId);
        SingleOutputStreamOperator<JSONObject> jsonObjStream = env.addSource(kafkaSource)
                .map(new MapFunction<String, JSONObject>() {
                    @Override
                    public JSONObject map(String s) throws Exception {
                        return JSON.parseObject(s);
                    }
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<JSONObject>
                                forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                    @Override
                                    public long extractTimestamp(JSONObject jsonObject, long l) {
                                        return jsonObject.getLong("ts");
                                    }
                                })
                );

        // 3. 提取流量来源信息并分类
        SingleOutputStreamOperator<TrafficSourceStats> sourceStatsStream = jsonObjStream.map(new MapFunction<JSONObject, TrafficSourceStats>() {
            @Override
            public TrafficSourceStats map(JSONObject jsonObject) throws Exception {
                String statsTime = DateFormatUtil.toYmdHms(jsonObject.getLong("ts"));                String statsType = "10s";
                
                // 提取来源信息
                JSONObject common = jsonObject.getJSONObject("common");
                String mid = common != null ? common.getString("mid") : "unknown";
                String ch = common != null ? common.getString("ch") : "unknown";
                
                // 安全地获取page对象和entry属性
                String entry = "unknown";
                JSONObject page = jsonObject.getJSONObject("page");
                if (page != null) {
                    // 由于直接从ods层读取，这里需要适配ods层的数据结构
                    // 使用page_id代替entry来进行分类
                    String pageId = page.getString("page_id");
                    if (pageId != null) {
                        entry = pageId;
                    }
                }
                
                // 确定来源类型和二级来源
                String sourceType = "other";
                String secondSource = "unknown";
                
                // 商品流量：通过商品详情页、商品搜索等进入
                if (entry != null && (entry.contains("goods") || entry.contains("search"))) {
                    sourceType = "商品流量";
                    secondSource = entry;
                }
                // 店铺流量：通过首页、分类页、活动页等进入
                else if (entry != null && (entry.equals("home") || entry.contains("category") || entry.contains("activity"))) {
                    sourceType = "店铺流量";
                    secondSource = entry;
                }
                // 直播流量：通过直播间进入
                else if (entry != null && entry.contains("live")) {
                    sourceType = "直播流量";
                    secondSource = entry;
                }
                // 内容流量：通过文章、视频等内容进入
                else if (entry != null && (entry.contains("article") || entry.contains("video"))) {
                    sourceType = "内容流量";
                    secondSource = entry;
                }

                // 初始化统计指标
                TrafficSourceStats stats = new TrafficSourceStats();
                stats.setStatsTime(statsTime);
                stats.setStatsType(statsType);
                stats.setSourceType(sourceType);
                stats.setSecondSource(secondSource);
                stats.setUvCount(1L); // 每个mid算作一个访客
                stats.setPvCount(1L); // 每个页面访问算作一个PV
                
                // 安全地获取page_id并计算goodsUvCount
                long goodsUvCount = 0L;
                if (page != null && "good_detail".equals(page.getString("page_id"))) {
                    goodsUvCount = 1L;
                }
                stats.setGoodsUvCount(goodsUvCount);
                
                // 实际项目中应从订单数据获取支付金额
                stats.setPaymentAmount(100.0); // 模拟值，实际应从订单数据中获取
                stats.setCreateTime(new Date());
                
                return stats;
            }
        });

        // 4. 按来源类型和二级来源分组，统计指标
        SingleOutputStreamOperator<TrafficSourceStats> aggregatedStream = sourceStatsStream
                .keyBy(new KeySelector<TrafficSourceStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(TrafficSourceStats stats) throws Exception {
                        return new Tuple3<>(stats.getStatsTime(), stats.getSourceType(), stats.getSecondSource());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .reduce(new ReduceFunction<TrafficSourceStats>() {
                    @Override
                    public TrafficSourceStats reduce(TrafficSourceStats value1, TrafficSourceStats value2) throws Exception {
                        // 累加各项指标
                        value1.setUvCount(value1.getUvCount() + value2.getUvCount());
                        value1.setPvCount(value1.getPvCount() + value2.getPvCount());
                        value1.setGoodsUvCount(value1.getGoodsUvCount() + value2.getGoodsUvCount());
                        value1.setPaymentAmount(value1.getPaymentAmount() + value2.getPaymentAmount());
                        return value1;
                    }
                }, new WindowFunction<TrafficSourceStats, TrafficSourceStats, Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple3<String, String, String> key, TimeWindow window,
                                     Iterable<TrafficSourceStats> input, Collector<TrafficSourceStats> out) throws Exception {
                        out.collect(input.iterator().next());
                    }
                });

        // 5. 按来源类型分组，计算TOP10
        SingleOutputStreamOperator<TrafficSourceStats> top10Stream = aggregatedStream
                .keyBy(new KeySelector<TrafficSourceStats, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> getKey(TrafficSourceStats stats) throws Exception {
                        // 按统计时间、统计类型、来源类型分组，对每个类型计算TOP10
                        return new Tuple3<>(stats.getStatsTime(), stats.getStatsType(), stats.getSourceType());
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
                .apply(new WindowFunction<TrafficSourceStats, TrafficSourceStats, Tuple3<String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple3<String, String, String> key, TimeWindow window,
                                     Iterable<TrafficSourceStats> input, Collector<TrafficSourceStats> out) throws Exception {
                        // 将窗口内的数据转换为列表
                        List<TrafficSourceStats> statsList = new ArrayList<>();
                        for (TrafficSourceStats stats : input) {
                            statsList.add(stats);
                        }

                        // 按访客数降序排序，取TOP10
                        List<TrafficSourceStats> top10List = statsList.stream()
                                .sorted(Comparator.comparing(TrafficSourceStats::getUvCount).reversed())
                                .limit(10)
                                .collect(Collectors.toList());

                        // 为TOP10数据添加排名
                        for (int i = 0; i < top10List.size(); i++) {
                            TrafficSourceStats stats = top10List.get(i);
                            stats.setRank(i + 1);
                            out.collect(stats);
                        }
                    }
                });

        // 6. 将结果写入ClickHouse
        top10Stream.addSink(MyClickHouseUtil.getSinkFunction(
                "insert into dws_traffic_source_topn_window values(?,?,?,?,?,?,?,?,?,?)"
        ));

        System.out.println("TrafficSourceTopNApp 已启动，正在从ods_traffic主题读取数据...");
        // 7. 执行作业
        env.execute("TrafficSourceTopNApp");
    }
}

